package com.davon.risk.controll;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.time.Duration;

public class DynamicCategoryLimit {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<OrderInfo> orderDS = env.socketTextStream("localhost", 9999, "\n", 3)
                .map(line -> {
                    String[] fields = line.split(",");
                    return new OrderInfo(Integer.valueOf(fields[0]), Integer.valueOf(fields[1]), Integer.valueOf(fields[1]), Integer.valueOf(fields[3]), Long.valueOf(fields[4]))
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(5L)).withTimestampAssigner(new SerializableTimestampAssigner<OrderInfo>() {
                    @Override
                    public long extractTimestamp(OrderInfo element, long recordTimestamp) {
                        return element.ts;
                    }
                }));

        SingleOutputStreamOperator<CategoryRule> ruleDs = env.socketTextStream("localhost", 9999, "\n", 3)
                .map(line -> {
                    String[] fields = line.split(",");
                    return new CategoryRule(fields[0], Integer.valueOf(fields[1]), Integer.valueOf(fields[2]));
                });

        MapStateDescriptor<Integer, CategoryRule> ruleState = new MapStateDescriptor<>("rule-map-state", Integer.class, CategoryRule.class);
        BroadcastStream<CategoryRule> ruleBroadcastDS = ruleDs.broadcast(ruleState);

        orderDS.connect(ruleBroadcastDS)
                .process(new MyCateRuleProcessFunction(ruleState))
                .keyBy(OrderWithRule::getCategoryId)
                .window(TumblingEventTimeWindows.of(Time.days(1)))
                .trigger()
    }

}

class MyCateRuleProcessFunction extends BroadcastProcessFunction<OrderInfo, CategoryRule, OrderWithRule> {
    public MapStateDescriptor<Integer, CategoryRule> ruleState;

    public MyCateRuleProcessFunction(MapStateDescriptor<Integer, CategoryRule> ruleState) {
        this.ruleState = ruleState;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        // 进行 jdbc 连接，将规则存入状态中
        Class.forName("com.mysql.jdbc.Drive");
        try {
            Connection conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test_demo?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC", "root", "password");
            Statement stmt = conn.createStatement();
            String sql = "select category_id, reward_limit from t_category_risk_limit where state = 1";
            ResultSet rs = stmt.executeQuery(sql);
            MapState<Integer, CategoryRule> rules = getRuntimeContext().getMapState(ruleState);
            while (rs.next()) {
                rules.put(rs.getInt("category_id"), new CategoryRule(rs.getString(""), rs.getInt(""), rs.getInt("")));
            }
            rs.close();
            stmt.close();
            conn.close();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    @Override
    public void processElement(OrderInfo value, ReadOnlyContext ctx, Collector<OrderWithRule> out) throws Exception {
        ReadOnlyBroadcastState<Integer, CategoryRule> rules = ctx.getBroadcastState(ruleState);
        Integer rewardLimit = null;
        if (rules != null) {
            CategoryRule cateRule = rules.get(value.categoryId);
            if (cateRule != null) {
                rewardLimit = cateRule.rewardLimit;
            }
        }
        out.collect(new OrderWithRule(value.orderId, value.categoryId, value.reward, rewardLimit));
    }

    @Override
    public void processBroadcastElement(CategoryRule value, Context ctx, Collector<OrderWithRule> out) throws Exception {
        BroadcastState<Integer, CategoryRule> rules = ctx.getBroadcastState(ruleState);
        if ("delete".equals(value.operate)) {
            rules.remove(value.categoryId);
        } else {
            rules.put(value.categoryId, value);
        }
    }
}

class FixedIntervalTrigger extends Trigger<OrderWithRule, Window> {
    private Long interval; // 触发时间间隔

    

    @Override
    public TriggerResult onElement(OrderWithRule element, long timestamp, Window window, TriggerContext ctx) throws Exception {

        return null;
    }

    @Override
    public TriggerResult onProcessingTime(long time, Window window, TriggerContext ctx) throws Exception {
        return null;
    }

    @Override
    public TriggerResult onEventTime(long time, Window window, TriggerContext ctx) throws Exception {
        return null;
    }

    @Override
    public void clear(Window window, TriggerContext ctx) throws Exception {

    }
}

class OrderInfo {
    public Integer orderId;
    public Integer categoryId;
    public Integer uid;
    public Integer reward;
    public Long ts;

    public OrderInfo(Integer orderId, Integer categoryId, Integer uid, Integer reward, Long ts) {
        this.orderId = orderId;
        this.categoryId = categoryId;
        this.uid = uid;
        this.reward = reward;
        this.ts = ts;
    }
}

class CategoryRule {
    public String operate;
    public Integer categoryId;
    public Integer rewardLimit;

    public CategoryRule(String operate, Integer categoryId, Integer rewardLimit) {
        this.operate = operate;
        this.categoryId = categoryId;
        this.rewardLimit = rewardLimit;
    }
}

class OrderWithRule {
    public Integer orderId;
    public Integer categoryId;
    public Integer reward;
    public Integer rewardLimit;

    public OrderWithRule(Integer orderId, Integer categoryId, Integer reward, Integer rewardLimit) {
        this.orderId = orderId;
        this.categoryId = categoryId;
        this.reward = reward;
        this.rewardLimit = rewardLimit;
    }

    public Integer getCategoryId() {
        return this.categoryId;
    }
}
